package day01;

import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.text.SimpleDateFormat;
import java.util.Arrays;

public class TestSource {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        //env.readTextFile()
        //env.readFile()
        //env.socketTextStream()
        //env.fromElements()
        //env.fromCollection()
        //env.generateSequence()
        env.readTextFile("data/a.txt").print();

        env.readFile(new TextInputFormat(new Path("data/a.txt")),"data/a.txt").print();



//        env.socketTextStream("localhost",6667).print();
        //参数 可变数组
//        env.fromElements(1,2,3,4,5,6).print();
////
//        env.fromCollection(Arrays.asList(1,2,3,4,5)).print();
//
//        env.generateSequence(0,100);


        //addSource
        env.addSource(new SourceFunction<String>() {

            Boolean isRunning=true;
            int interval= 1000;
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                int id=0;
                while (isRunning){
                    ctx.collect(id+"TIME"+new SimpleDateFormat("yyyy年MM月dd日 HH时mm分ss秒SSS毫秒").format(System.currentTimeMillis()));
                    Thread.sleep(interval);
                    id++;
                }
            }
            //web cancel退出执行
            @Override
            public void cancel() {
                System.out.println("webJob关闭");
                isRunning=false;
            }
        }).print();

        env.execute();

        System.out.println("git_test");
        System.out.println("master");
        System.out.println("hot-fix");
        System.out.println("gitee-push");
        System.out.println("gitee-pull");
        

    }
}
